package org.weibo.analysis.database;


import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.weibo.analysis.entity.Vertex;
import org.weibo.analysis.graph.Edge;
import org.weibo.analysis.graph.GraphContainer;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Map;

public class SqliteSink extends RichSinkFunction<GraphContainer> {

    private static final Logger logger = LoggerFactory.getLogger("SqliteSink");
    private Connection conn;
    private PreparedStatement ps;


    @Override
    public void invoke(GraphContainer graphContainer, Context context) throws Exception {
        try {
            ps = conn.prepareStatement("insert into weibo_link_data(id, label,source,target,remark,timestamp,count) values(?,?,?,?,?,?,?)");
            for (Map.Entry<String, Edge<Vertex, Vertex>> entry : graphContainer.getEdges().entrySet()) {
                Edge<Vertex, Vertex> value = entry.getValue();

                ps.setString(1, value.id);
                ps.setString(2, value.label);
                ps.setString(3, value.source.toJson());
                ps.setString(4, value.target.toJson());
                ps.setString(5, value.remark);
                ps.setLong(6, value.timestamp);
                ps.setInt(7, value.count);
                ps.addBatch();

            }
            ps.executeBatch();
            conn.commit();
            graphContainer.clear();
        } catch (SQLException throwable) {
            throwable.printStackTrace();
            logger.error("record sunk failed：" + graphContainer.getEdges().keySet().toString());
        }
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        try {
            Class.forName(Config.SQLITE_DRIVER_NAME);
            conn = DriverManager.getConnection(Config.SQLITE_JDBC_URL);
            conn.setAutoCommit(false);
        } catch (ClassNotFoundException | SQLException e) {
            e.printStackTrace();
            logger.error("init mysql error");
        }
    }

    @Override
    public void close() throws Exception {

        if (conn != null) {
            try {
                conn.commit();
                conn.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }

        }
    }

}